package com.g2room.bigdata.hadoop.hdfs;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.*;
import org.apache.hadoop.io.IOUtils;
import org.apache.hadoop.util.Progressable;

import java.io.*;
import java.util.ArrayList;
import java.util.List;
import java.util.Optional;

/**
 * HDFS工具类，实现HDFS的相关操作：列表、创建、上传、下载
 * FileSystem：比较通用的文件系统的抽象基类，可以实现分布式系统或者本地连接磁盘，本地版本可以做hadoop的小测试。
 * 要想使用HDFS，那么就得使用这个FileSystem的对象或者子类对象。
 * 本地实现是 LocalFileSystem，分布式实现是 DistributedFileSystem。剩下的还有其他第三方实现。
 */
public class HdfsUtil {
    /**
     * 配置，此时会去加载jar包里面的 hdfs-default.xml
     */
    static Configuration conf = new Configuration(true);

    static {
        //指定hadoop fs的地址，后面改为配置
        conf.set("fs.default.name", "hdfs://192.168.0.212:9000");
    }

    public static FileSystem initFs() throws IOException {
        // 根据conf的fs.default.name判断是具体实例化 LocalFileSystem 还是 DistributedFileSystem
        // 如果未配置，则从工程的classpath读取，也读取不到再从hadoop的jar包里面的core-default.xml读取。默认值为：file://
        return FileSystem.get(conf);
    }

    public static void closeFs(FileSystem fs) {
        try {
            if (null != fs) {
                fs.close();
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    /**
     * 获取hdfs路径下的文件列表
     *
     * @param path HDFS文件路径
     * @return
     */
    public static List<HdfsFileTo> getFileList(String path) {
        FileSystem fs = null;
        List<HdfsFileTo> files = new ArrayList<HdfsFileTo>();
        try {
            fs = initFs();
            Path origin = new Path(path);
            // 只列目录下一层的文件夹或文件
            FileStatus[] status = fs.listStatus(origin);
            for(FileStatus fileStatus :status) {
                HdfsFileTo hdfsFile = new HdfsFileTo();
                hdfsFile.setBlockSize(fileStatus.getBlockSize());
                hdfsFile.setGroup(fileStatus.getGroup());
                hdfsFile.setLastModified(fileStatus.getModificationTime());
                hdfsFile.setName(fileStatus.getPath().getName());
                hdfsFile.setOwner(fileStatus.getOwner());
                hdfsFile.setPermission(fileStatus.getPermission());
                hdfsFile.setReplication(fileStatus.getReplication());
                hdfsFile.setSize(fileStatus.getLen());
                hdfsFile.setDir(fileStatus.isDirectory());
                files.add(hdfsFile);
            }
            /**
             RemoteIterator<LocatedFileStatus> status = fs.listLocatedStatus(origin);
             // 遍历指定目录下的所有文件（不列出文件夹），遍历列出文件
             //            RemoteIterator<LocatedFileStatus> status = fs.listFiles(origin, false);
             while (status.hasNext()) {
             HdfsFileTo hdfsFile = new HdfsFileTo();
             LocatedFileStatus fileStatus = status.next();
             hdfsFile.setBlockSize(fileStatus.getBlockSize());
             hdfsFile.setGroup(fileStatus.getGroup());
             hdfsFile.setLastModified(fileStatus.getModificationTime());
             hdfsFile.setName(fileStatus.getPath().getName());
             hdfsFile.setOwner(fileStatus.getOwner());
             hdfsFile.setPermission(fileStatus.getPermission());
             hdfsFile.setReplication(fileStatus.getReplication());
             hdfsFile.setSize(fileStatus.getLen());
             hdfsFile.setDir(fileStatus.isDirectory());
             files.add(hdfsFile);
             }
             */
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            closeFs(fs);
        }
        return files;
    }

    /**
     * 创建HDFS文件夹
     *
     * @param path HDFS文件夹路径
     * @return 创建成功返回 true，失败则返回false
     */
    public static boolean mkdir(String path) {
        FileSystem fs = null;
        boolean rs = false;
        try {
            fs = initFs();
            rs = fs.mkdirs(new Path(path));
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            closeFs(fs);
        }
        return rs;
    }

    /**
     * 创建HDFS文件
     *
     * @param path HDFS文件路径
     * @return 创建成功返回 true，失败则返回false
     */
    public static Optional<FSDataOutputStream> create(String path) {
        FileSystem fs = null;
        FSDataOutputStream outputStream = null;
        try {
            fs = initFs();
            // 后面的true表示如果文件存在，则覆盖
            outputStream = fs.create(new Path(path), true);
        } catch (Exception e) {
            e.printStackTrace();
//        } finally {
//            closeFs(fs);
        }
        return Optional.of(outputStream);
    }

    /**
     * 删除HDFS文件
     *
     * @param path HDFS文件路径
     * @return 删除成功返回 true，失败则返回false
     */
    public static boolean delete(String path) {
        FileSystem fs = null;
        boolean rs = false;
        try {
            fs = initFs();
            // 后面的true表示是否递归删除
            rs = fs.delete(new Path(path), true);
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            closeFs(fs);
        }
        return rs;
    }

    /**
     * 文件或文件夹重命名
     *
     * @param sourcePath 源文件或文件夹路径（待修改的）
     * @param targetPath 目标文件或文件夹路径（修改后的）
     * @return 删除成功返回 true，失败则返回false
     */
    public static boolean delete(String sourcePath, String targetPath) {
        FileSystem fs = null;
        boolean rs = false;
        try {
            fs = initFs();
            rs = fs.rename(new Path(sourcePath), new Path(targetPath));
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            closeFs(fs);
        }
        return rs;
    }

    /**
     * HDFS文件写入内容，此处没有关闭写入流，需要在调用的方法里面关闭
     *
     * @param content 待写入的内容
     * @return 创建成功返回 true，失败则返回false
     */
    public static void write(FSDataOutputStream outputStream, String content) {
        try {
            outputStream.write(content.getBytes());
            outputStream.flush();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    /**
     * 查看HDFS文件的内容
     *
     * @param path 待读取的文件路径
     * @return 创建成功返回 true，失败则返回false
     */
    public static void cat(String path) {
        FileSystem fs = null;
        FSDataInputStream inputStream = null;
        try {
            fs = initFs();
            inputStream = fs.open(new Path(path));
            // 内容输出到控制台
            IOUtils.copyBytes(inputStream, System.out, 1024);
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            try {
                if (null != inputStream) {
                    inputStream.close();
                }
            } catch (IOException e) {
                e.printStackTrace();
            }
            closeFs(fs);
        }
    }

    /**
     * 上传小文件到HDFS
     *
     * @param localPath 本地文件路径
     * @param hdfsPath  HDFS文件路径
     */
    public static void upload(String localPath, String hdfsPath) {
        FileSystem fs = null;
        try {
            fs = initFs();
            fs.copyFromLocalFile(new Path(localPath), new Path(hdfsPath));
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            closeFs(fs);
        }
    }

    /**
     * 上传本地大文件到HDFS，带进度条
     *
     * @param localPath 本地文件路径
     * @param hdfsPath  HDFS文件路径
     */
    public static void uploadWithProgress(String localPath, String hdfsPath) {
        FileSystem fs = null;
        InputStream inputStream = null;
        FSDataOutputStream outputStream = null;
        try {
            fs = initFs();
            inputStream = new BufferedInputStream(new FileInputStream(new File(localPath)));
            outputStream = fs.create(new Path(hdfsPath), new Progressable() {
                @Override
                public void progress() {
                    // 这里处理进度信息
                    System.out.println(".");
                }
            });
            IOUtils.copyBytes(inputStream, outputStream, 4096);
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            try {
                if (null != outputStream) {
                    outputStream.close();
                }
            } catch (IOException e) {
                e.printStackTrace();
            }
            try {
                if (null != inputStream) {
                    inputStream.close();
                }
            } catch (IOException e) {
                e.printStackTrace();
            }
            closeFs(fs);
        }
    }

    /**
     * 下载HDFS文件到本地
     *
     * @param localPath 本地文件路径
     * @param hdfsPath  HDFS文件路径
     */
    public static void download(String localPath, String hdfsPath) {
        FileSystem fs = null;
        try {
            fs = initFs();
            fs.copyToLocalFile(new Path(hdfsPath), new Path(localPath));
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            closeFs(fs);
        }
    }
}
